package zhongche.source;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class MySourceFunction implements SourceFunction<JSONObject> {

    private Boolean isRunning;

    private JSONObject jsonObject;

    private Long startDateTime;

    private Integer speed;

    public MySourceFunction(Long startDateTime, Integer speed) {
        this.speed = speed;
        this.startDateTime = startDateTime;
        this.jsonObject = JSONObject.parseObject("{\n" +
                "  \"box_num\": 1,\n" +
                "  \"current_station\": 5,\n" +
                "  \"dDevice\": 50,\n" +
                "  \"dataParam\": {\n" +
                "\"s0000108\": 0,\n" +
                "\"s0915691\": " + speed + ",\n" +
                "    \"s1999915\": 1\n" +
                "  },\n" +
                "  \"data_sDeviceNo\": 20,\n" +
                "  \"frame_num\": 5657,\n" +
                "  \"line_id\": 2,\n" +
                "  \"line_num\": \"2\",\n" +
                "  \"message_type\": 5025,\n" +
                "  \"netLink_flag\": 0,\n" +
                "  \"next_station\": 0,\n" +
                "  \"ocs_time\": \"210716153627\",\n" +
                "  \"sDevice\": 40,\n" +
                "  \"sDeviceTime\": \"" + startDateTime + "\",   \n" +
                "  \"second_circle\": 1,\n" +
                "  \"speed\": 0.0,\n" +
                "  \"start_station\": 5,\n" +
                "  \"terminal_station\": 28,\n" +
                "  \"train_id\": 159160,\n" +
                "  \"train_name\": \"159160\",\n" +
                "  \"train_num\": 159160,\n" +
                "  \"train_type\": \"A52\",\n" +
                "  \"ttype_id\": 10\n" +
                "}\n");
    }

    @Override
    public void run(SourceContext<JSONObject> ctx) throws Exception {
        while (isRunning){
            ctx.collect(this.jsonObject);
            Thread.sleep(1000L);
            this.speed -= 10;
            if(this.speed < 0){
                this.speed = 0;
            }

        }

    }

    @Override
    public void cancel() {

    }
}
